package fun.tan90.device.source.mysql.plugin.config;

import cn.hutool.core.lang.Pair;
import cn.hutool.core.util.ReUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import fun.tan90.device.event.model.DeviceSourceData;
import fun.tan90.device.source.core.SourceFactory;
import fun.tan90.device.source.mysql.plugin.model.BinlogClient;
import fun.tan90.device.source.mysql.plugin.model.BinlogPosition;
import fun.tan90.device.source.mysql.plugin.model.ColumnMetadata;
import fun.tan90.device.source.mysql.plugin.model.RowMutationEventData;
import fun.tan90.device.source.mysql.plugin.util.BinlogUtil;
import lombok.extern.slf4j.Slf4j;
import org.noear.snack.ONode;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import org.redisson.api.RedissonClient;

import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.BiConsumer;

@Slf4j
@Component
public class MysqlClientFactory extends SourceFactory {

    @Inject
    RedissonClient redissonClient;

    @Override
    public Class<?> init(ONode connectNode, BiConsumer<String, Object> biConsumer) {
        for (ONode node : connectNode.ary()) {
            String name = node.get("name").getString();
            ONode connect = node.get("connect");
            String hostname = connect.get("hostname").getString();
            int port = connect.get("port").getInt();
            String username = connect.get("username").getString();
            String password = connect.get("password").getString();
            String driverClassName = connect.get("driverClassName").getString();
            long serverId = connect.get("serverId").getLong();

            String str = redissonClient.getMap("binlog_position").getOrDefault(name, "").toString();
            BinlogPosition binlogPosition = StrUtil.isNotBlank(str) ? JSONUtil.toBean(str, BinlogPosition.class) : new BinlogPosition();
            BinlogClient binlogClient = new BinlogClient(hostname, port, username, password, driverClassName, serverId, binlogPosition.getPosition(), binlogPosition.getFilename());
            biConsumer.accept(name, binlogClient);
        }
        return BinlogClient.class;
    }

    @Override
    public void post(Map<String, ONode> ruleMap, Map<String, ?> clientMap) {
        for (String name : clientMap.keySet()) {
            String database = ruleMap.get(name).select("$.config.database").getString();
            String table = ruleMap.get(name).select("$.config.table").getString();
            String deviceCodeColName = ruleMap.get(name).select("$.config.deviceCodeColName").getString();
            long timeOffset = ruleMap.get(name).select("$.config.timeOffset").getLong();
            if (clientMap.get(name) instanceof BinlogClient binlogClient) {
                binlogClient.registerEventListener(event -> {
                    EventHeaderV4 header = event.getHeader();
                    EventType eventType = header.getEventType();
                    EventData eventData = event.getData();
                    //log.info("name {} event_type {}", name, eventType);
                    if (eventType == EventType.TABLE_MAP) {
                        if (eventData instanceof TableMapEventData tableMapEventData) {
                            boolean databaseMatch = ReUtil.isMatch(database, tableMapEventData.getDatabase());
                            boolean tableMatch = ReUtil.isMatch(table, tableMapEventData.getTable());
                            if (databaseMatch && tableMatch) {
                                binlogClient.putTablePair(tableMapEventData.getTableId(), Pair.of(tableMapEventData.getDatabase(), tableMapEventData.getTable()));
                            }
                        }
                    } else {
                        if (EventType.isRowMutation(eventType)) {
                            RowMutationEventData rowMutationEventData = new RowMutationEventData(eventData);
                            Pair<String, String> pair = binlogClient.getTablePair(rowMutationEventData.getTableId());
                            if (Objects.nonNull(pair)) {
                                List<ColumnMetadata> columnMetadata = binlogClient.columnMetadata(pair.getKey(), pair.getValue());
                                if (EventType.isUpdate(eventType)) {
                                    List<Map.Entry<Serializable[], Serializable[]>> updateRows = rowMutationEventData.getUpdateRows();
                                    for (Map.Entry<Serializable[], Serializable[]> updateRow : updateRows) {
                                        ONode updateNode = BinlogUtil.toNode(columnMetadata, updateRow.getValue(), timeOffset);
                                        log.info("【{}】=> update_row {}", name, updateNode.toJson());
                                        afterPost(new DeviceSourceData(database + "." + table + ".update", updateNode.get(deviceCodeColName).getString(), updateNode.toJson()));
                                    }
                                } else if (EventType.isDelete(eventType)) {
                                    List<Serializable[]> deleteRows = rowMutationEventData.getDeleteRows();
                                    for (Serializable[] deleteRow : deleteRows) {
                                        ONode deleteNode = BinlogUtil.toNode(columnMetadata, deleteRow, timeOffset);
                                        log.info("【{}】=> delete_row {}", name, deleteNode.toJson());
                                    }
                                } else if (EventType.isWrite(eventType)) {
                                    List<Serializable[]> insertRows = rowMutationEventData.getInsertRows();
                                    for (Serializable[] insertRow : insertRows) {
                                        ONode insertNode = BinlogUtil.toNode(columnMetadata, insertRow, timeOffset);
                                        log.info("【{}】=> insert_row {}", name, insertNode.toJson());
                                        afterPost(new DeviceSourceData(database + "." + table + ".insert", insertNode.get(deviceCodeColName).getString(), insertNode.toJson()));
                                    }
                                }
                            }
                        } else {
                            if (eventType != EventType.FORMAT_DESCRIPTION) {
                                if (EventType.ROTATE == eventType) {
                                    if (eventData instanceof RotateEventData rotateEventData) {
                                        redissonClient.getMap("binlog_position").put(name, JSONUtil.toJsonStr(BinlogPosition.builder().position(rotateEventData.getBinlogPosition()).filename(rotateEventData.getBinlogFilename()).build()));
                                    }
                                } else {
                                    String str = redissonClient.getMap("binlog_position").getOrDefault(name, "").toString();
                                    if (StrUtil.isNotBlank(str)) {
                                        BinlogPosition binlogPosition = JSONUtil.toBean(str, BinlogPosition.class);
                                        binlogPosition.setPosition(header.getNextPosition());
                                        redissonClient.getMap("binlog_position").put(name, JSONUtil.toJsonStr(binlogPosition));
                                    }
                                }
                            }
                        }
                    }
                });

                binlogClient.registerLifecycleListener(new BinaryLogClient.LifecycleListener() {
                    @Override
                    public void onConnect(BinaryLogClient client) {
                        log.info("【{}】连接成功", name);
                        clientInitSuccess(name);
                    }

                    @Override
                    public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
                        log.error("【{}】onCommunicationFailure", name, ex);
                        clientInitFailure(name, ex);
                    }

                    @Override
                    public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
                        log.error("【{}】onEventDeserializationFailure", name, ex);
                    }

                    @Override
                    public void onDisconnect(BinaryLogClient client) {
                        log.info("【{}】onDisconnect", name);
                    }
                });

                binlogClient.connect(redissonClient, name, e -> clientInitFailure(name, e));
            }
        }
    }
}
